package com.dh.cb;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.stream.IntStream;

public class MessageConsumer {

    //消费数据
    public void start() {

        ConsumereThread consumereThread = new ConsumereThread();

        CyclicBarrier barrier = new CyclicBarrier(Constant.LOAD_THREAD, () -> consumereThread.saveOffset());

        List<LoadHandler> loadHandlers = new ArrayList<>(Constant.LOAD_THREAD);
        IntStream.range(0, Constant.LOAD_THREAD).forEach(i -> {
            LoadHandler loadHandler = new LoadHandler();
            LoadThread loadThread = new LoadThread(loadHandler, barrier);
            loadThread.setName("thread--" + i);
            loadThread.start();
            loadHandlers.add(loadHandler);
        });
        consumereThread.setHandlers(loadHandlers);
        consumereThread.start();
    }
}
